package com.imu.purse.mq.conumer;

import com.alibaba.fastjson.JSON;
import com.imu.purse.modules.bitcoin.YtfcoinApi;
import com.imu.purse.modules.trade.form.TransForm;
import com.imu.purse.modules.trade.service.TradeInfoService;
import com.imu.purse.mq.produce.YtfTransProduce;
import com.rabbitmq.client.Channel;
import com.xiaoleilu.hutool.log.Log;
import com.xiaoleilu.hutool.log.LogFactory;
import org.apache.commons.lang.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Component
public class YtfTransConsumer {

    private static final Log log = LogFactory.get(YtfTransConsumer.class);

    @Autowired
    private YtfcoinApi ytfcoinApi;

    @Autowired
    private YtfTransProduce ytfTransProduce;

    @Autowired
    private TradeInfoService tradeInfoService;


    /* DIRECT模式.
     * @param message the message
     * @param channel the channel
     */
    @RabbitListener(queues = {"YTF_TRANS_OUT_QUEUE"})
    public void message(Message message, Channel channel) {
        try{
            //解析消息
            parseData(message);
        }catch(Exception e){//抛出异常，则该条消息需要重新存入数据库中
            log.error(e);
            //解析消息
            sendToProduce(message);
        }finally {
            try{
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            }catch (Exception e){
                //解析消息
                sendToProduce(message);
                log.error(e);
            }
        }
    }

    /**
     * 如果有问题重新丢入队列中
     * @param message
     */
    private void sendToProduce(Message message){
        if(message != null && message.getBody() != null){
            String data = new String(message.getBody());
            if(StringUtils.isNotEmpty(data)){
                //将消息转换为对象
                TransForm bean = JSON.parseObject(data, TransForm.class);
                if(bean == null){
                    return;
                }
                ytfTransProduce.sendData(bean);
            }
        }
    }

    /**
     * 分析数据
     * @param message
     */
    private void parseData(Message message){

        if(message != null && message.getBody() != null){
            String data = new String(message.getBody());
            if(StringUtils.isNotEmpty(data)){
                //将消息转换为对象
                TransForm bean = JSON.parseObject(data, TransForm.class);
                if(bean == null){
                    return;
                }
                String transId = ytfcoinApi.sendTransaction(bean.getPayAddr(), bean.getRecAddr(),
                        bean.getMoney().toString(),bean.getPayPhone(), bean.getCoreAddr(),bean.getYtfType());
                if(transId == null){
                    sleepTime(10L);
                    ytfTransProduce.sendData(bean);
                }else{
                    bean.setTransId(transId);
                    tradeInfoService.transOutData(bean);
                    sleepTime(1L);
                }

            }
        }
    }
    /**
     * 休眠制定的时间
     * @param sleepTime
     */
    private void sleepTime(long sleepTime){
        try {
            TimeUnit.SECONDS.sleep(sleepTime);
        } catch (InterruptedException e) {
            log.error(e);
        }
    }
}
